{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "provenance": []
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "cells": [
    {
      "cell_type": "code",
      "source": [
        "import asyncio\n",
        "import json\n",
        "import os\n",
        "import yaml\n",
        "from pathlib import Path\n",
        "from typing import Any, Dict, Optional, Type, TypeVar, Union, get_type_hints\n",
        "from dataclasses import dataclass, field, MISSING\n",
        "from watchdog.observers import Observer\n",
        "from watchdog.events import FileSystemEventHandler\n",
        "import logging\n",
        "\n",
        "__version__ = \"0.1.0\"\n",
        "__author__ = \"AsyncConfig Team\"\n",
        "\n",
        "T = TypeVar('T')\n",
        "\n",
        "logger = logging.getLogger(__name__)"
      ],
      "metadata": {
        "id": "pSZnNbFldIrK"
      },
      "execution_count": 9,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "class ConfigError(Exception):\n",
        "    \"\"\"Base exception for configuration errors.\"\"\"\n",
        "    pass\n",
        "\n",
        "\n",
        "class ValidationError(ConfigError):\n",
        "    \"\"\"Raised when configuration validation fails.\"\"\"\n",
        "    pass\n",
        "\n",
        "\n",
        "class LoadError(ConfigError):\n",
        "    \"\"\"Raised when configuration loading fails.\"\"\"\n",
        "    pass\n",
        "\n",
        "\n",
        "@dataclass\n",
        "class ConfigSource:\n",
        "    \"\"\"Represents a configuration source with priority and reload capabilities.\"\"\"\n",
        "    path: Optional[Path] = None\n",
        "    env_prefix: Optional[str] = None\n",
        "    data: Optional[Dict[str, Any]] = None\n",
        "    priority: int = 0\n",
        "    watch: bool = False\n",
        "\n",
        "    def __post_init__(self):\n",
        "        if self.path:\n",
        "            self.path = Path(self.path)"
      ],
      "metadata": {
        "id": "HsU0EQbvfyqD"
      },
      "execution_count": 10,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "class ConfigWatcher(FileSystemEventHandler):\n",
        "    \"\"\"File system event handler for configuration hot reloading.\"\"\"\n",
        "\n",
        "    def __init__(self, config_manager, paths: list[Path]):\n",
        "        self.config_manager = config_manager\n",
        "        self.paths = {str(p.resolve()) for p in paths}\n",
        "        super().__init__()\n",
        "\n",
        "    def on_modified(self, event):\n",
        "        if not event.is_directory and event.src_path in self.paths:\n",
        "            logger.info(f\"Configuration file changed: {event.src_path}\")\n",
        "            asyncio.create_task(self.config_manager._reload_config())"
      ],
      "metadata": {
        "id": "q7od3CiUf1LF"
      },
      "execution_count": 11,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "class AsyncConfigManager:\n",
        "    \"\"\"\n",
        "    Modern async configuration manager with type safety and hot reloading.\n",
        "\n",
        "    Features:\n",
        "    - Async-first design\n",
        "    - Type-safe configuration classes\n",
        "    - Environment variable support\n",
        "    - Hot reloading\n",
        "    - Multiple source merging\n",
        "    - Validation with detailed error messages\n",
        "    \"\"\"\n",
        "\n",
        "    def __init__(self):\n",
        "        self.sources: list[ConfigSource] = []\n",
        "        self.observers: list[Observer] = []\n",
        "        self.config_cache: Dict[str, Any] = {}\n",
        "        self.reload_callbacks: list[callable] = []\n",
        "        self._lock = asyncio.Lock()\n",
        "\n",
        "    def add_source(self, source: ConfigSource) -> \"AsyncConfigManager\":\n",
        "        \"\"\"Add a configuration source.\"\"\"\n",
        "        self.sources.append(source)\n",
        "        self.sources.sort(key=lambda x: x.priority, reverse=True)\n",
        "        return self\n",
        "\n",
        "    def add_file(self, path: Union[str, Path], priority: int = 0, watch: bool = False) -> \"AsyncConfigManager\":\n",
        "        \"\"\"Add a file-based configuration source.\"\"\"\n",
        "        return self.add_source(ConfigSource(path=path, priority=priority, watch=watch))\n",
        "\n",
        "    def add_env(self, prefix: str, priority: int = 100) -> \"AsyncConfigManager\":\n",
        "        \"\"\"Add environment variable source.\"\"\"\n",
        "        return self.add_source(ConfigSource(env_prefix=prefix, priority=priority))\n",
        "\n",
        "    def add_dict(self, data: Dict[str, Any], priority: int = 50) -> \"AsyncConfigManager\":\n",
        "        \"\"\"Add dictionary-based configuration source.\"\"\"\n",
        "        return self.add_source(ConfigSource(data=data, priority=priority))\n",
        "\n",
        "    async def load_config(self, config_class: Type[T]) -> T:\n",
        "        \"\"\"Load and validate configuration into a typed dataclass.\"\"\"\n",
        "        async with self._lock:\n",
        "            config_data = await self._merge_sources()\n",
        "\n",
        "            try:\n",
        "                return self._validate_and_convert(config_data, config_class)\n",
        "            except Exception as e:\n",
        "                raise ValidationError(f\"Failed to validate configuration: {e}\")\n",
        "\n",
        "    async def _merge_sources(self) -> Dict[str, Any]:\n",
        "        \"\"\"Merge configuration from all sources based on priority.\"\"\"\n",
        "        merged = {}\n",
        "\n",
        "        for source in reversed(self.sources):\n",
        "            try:\n",
        "                data = await self._load_source(source)\n",
        "                if data:\n",
        "                    merged.update(data)\n",
        "            except Exception as e:\n",
        "                logger.warning(f\"Failed to load source {source}: {e}\")\n",
        "\n",
        "        return merged\n",
        "\n",
        "    async def _load_source(self, source: ConfigSource) -> Optional[Dict[str, Any]]:\n",
        "        \"\"\"Load data from a single configuration source.\"\"\"\n",
        "        if source.data:\n",
        "            return source.data.copy()\n",
        "\n",
        "        if source.path:\n",
        "            return await self._load_file(source.path)\n",
        "\n",
        "        if source.env_prefix:\n",
        "            return self._load_env_vars(source.env_prefix)\n",
        "\n",
        "        return None\n",
        "\n",
        "    async def _load_file(self, path: Path) -> Dict[str, Any]:\n",
        "        \"\"\"Load configuration from a file.\"\"\"\n",
        "        if not path.exists():\n",
        "            raise LoadError(f\"Configuration file not found: {path}\")\n",
        "\n",
        "        try:\n",
        "            content = await asyncio.to_thread(path.read_text)\n",
        "\n",
        "            if path.suffix.lower() == '.json':\n",
        "                return json.loads(content)\n",
        "            elif path.suffix.lower() in ['.yml', '.yaml']:\n",
        "                return yaml.safe_load(content) or {}\n",
        "            else:\n",
        "                raise LoadError(f\"Unsupported file format: {path.suffix}\")\n",
        "\n",
        "        except Exception as e:\n",
        "            raise LoadError(f\"Failed to load {path}: {e}\")\n",
        "\n",
        "    def _load_env_vars(self, prefix: str) -> Dict[str, Any]:\n",
        "        \"\"\"Load environment variables with given prefix.\"\"\"\n",
        "        env_vars = {}\n",
        "        prefix = prefix.upper() + '_'\n",
        "\n",
        "        for key, value in os.environ.items():\n",
        "            if key.startswith(prefix):\n",
        "                config_key = key[len(prefix):].lower()\n",
        "                env_vars[config_key] = self._convert_env_value(value)\n",
        "\n",
        "        return env_vars\n",
        "\n",
        "    def _convert_env_value(self, value: str) -> Any:\n",
        "        \"\"\"Convert environment variable string to appropriate type.\"\"\"\n",
        "        if value.lower() in ('true', 'false'):\n",
        "            return value.lower() == 'true'\n",
        "\n",
        "        try:\n",
        "            if '.' in value:\n",
        "                return float(value)\n",
        "            return int(value)\n",
        "        except ValueError:\n",
        "            pass\n",
        "\n",
        "        try:\n",
        "            return json.loads(value)\n",
        "        except json.JSONDecodeError:\n",
        "            pass\n",
        "\n",
        "        return value\n",
        "\n",
        "    def _validate_and_convert(self, data: Dict[str, Any], config_class: Type[T]) -> T:\n",
        "        \"\"\"Validate and convert data to the specified configuration class.\"\"\"\n",
        "        if not hasattr(config_class, '__dataclass_fields__'):\n",
        "            raise ValidationError(f\"{config_class.__name__} must be a dataclass\")\n",
        "\n",
        "        type_hints = get_type_hints(config_class)\n",
        "        field_values = {}\n",
        "\n",
        "        for field_name, field_info in config_class.__dataclass_fields__.items():\n",
        "            if field_name in data:\n",
        "                field_value = data[field_name]\n",
        "\n",
        "                if hasattr(field_info.type, '__dataclass_fields__'):\n",
        "                    if isinstance(field_value, dict):\n",
        "                        field_value = self._validate_and_convert(field_value, field_info.type)\n",
        "\n",
        "                field_values[field_name] = field_value\n",
        "            elif field_info.default is not MISSING:\n",
        "                field_values[field_name] = field_info.default\n",
        "            elif field_info.default_factory is not MISSING:\n",
        "                field_values[field_name] = field_info.default_factory()\n",
        "            else:\n",
        "                raise ValidationError(f\"Required field '{field_name}' not found in configuration\")\n",
        "\n",
        "        return config_class(**field_values)\n",
        "\n",
        "    async def start_watching(self):\n",
        "        \"\"\"Start watching configuration files for changes.\"\"\"\n",
        "        watch_paths = []\n",
        "\n",
        "        for source in self.sources:\n",
        "            if source.watch and source.path:\n",
        "                watch_paths.append(source.path)\n",
        "\n",
        "        if watch_paths:\n",
        "            observer = Observer()\n",
        "            watcher = ConfigWatcher(self, watch_paths)\n",
        "\n",
        "            for path in watch_paths:\n",
        "                observer.schedule(watcher, str(path.parent), recursive=False)\n",
        "\n",
        "            observer.start()\n",
        "            self.observers.append(observer)\n",
        "            logger.info(f\"Started watching {len(watch_paths)} configuration files\")\n",
        "\n",
        "    async def stop_watching(self):\n",
        "        \"\"\"Stop watching configuration files.\"\"\"\n",
        "        for observer in self.observers:\n",
        "            observer.stop()\n",
        "            observer.join()\n",
        "        self.observers.clear()\n",
        "\n",
        "    async def _reload_config(self):\n",
        "        \"\"\"Reload configuration from all sources.\"\"\"\n",
        "        try:\n",
        "            self.config_cache.clear()\n",
        "            for callback in self.reload_callbacks:\n",
        "                await callback()\n",
        "            logger.info(\"Configuration reloaded successfully\")\n",
        "        except Exception as e:\n",
        "            logger.error(f\"Failed to reload configuration: {e}\")\n",
        "\n",
        "    def on_reload(self, callback: callable):\n",
        "        \"\"\"Register a callback to be called when configuration is reloaded.\"\"\"\n",
        "        self.reload_callbacks.append(callback)\n",
        "\n",
        "    async def __aenter__(self):\n",
        "        await self.start_watching()\n",
        "        return self\n",
        "\n",
        "    async def __aexit__(self, exc_type, exc_val, exc_tb):\n",
        "        await self.stop_watching()"
      ],
      "metadata": {
        "id": "w9Ugybilf2Rg"
      },
      "execution_count": 12,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "async def load_config(config_class: Type[T],\n",
        "                     config_file: Optional[Union[str, Path]] = None,\n",
        "                     env_prefix: Optional[str] = None,\n",
        "                     watch: bool = False) -> T:\n",
        "    \"\"\"\n",
        "    Convenience function to quickly load configuration.\n",
        "\n",
        "    Args:\n",
        "        config_class: Dataclass to load configuration into\n",
        "        config_file: Optional configuration file path\n",
        "        env_prefix: Optional environment variable prefix\n",
        "        watch: Whether to watch for file changes\n",
        "\n",
        "    Returns:\n",
        "        Configured instance of config_class\n",
        "    \"\"\"\n",
        "    manager = AsyncConfigManager()\n",
        "\n",
        "    if config_file:\n",
        "        manager.add_file(config_file, priority=0, watch=watch)\n",
        "\n",
        "    if env_prefix:\n",
        "        manager.add_env(env_prefix, priority=100)\n",
        "\n",
        "    return await manager.load_config(config_class)"
      ],
      "metadata": {
        "id": "9bpA9vS6gGNq"
      },
      "execution_count": 13,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "@dataclass\n",
        "class DatabaseConfig:\n",
        "    \"\"\"Example database configuration.\"\"\"\n",
        "    host: str = \"localhost\"\n",
        "    port: int = 5432\n",
        "    username: str = \"admin\"\n",
        "    password: str = \"\"\n",
        "    database: str = \"myapp\"\n",
        "    ssl_enabled: bool = False\n",
        "    pool_size: int = 10\n",
        "\n",
        "\n",
        "@dataclass\n",
        "class AppConfig:\n",
        "    \"\"\"Example application configuration.\"\"\"\n",
        "    debug: bool = False\n",
        "    log_level: str = \"INFO\"\n",
        "    secret_key: str = \"\"\n",
        "    database: DatabaseConfig = field(default_factory=DatabaseConfig)\n",
        "    redis_url: str = \"redis://localhost:6379\"\n",
        "    max_workers: int = 4\n",
        "\n",
        "\n",
        "async def demo_simple_config():\n",
        "    \"\"\"Demo simple configuration loading.\"\"\"\n",
        "\n",
        "    sample_config = {\n",
        "        \"debug\": True,\n",
        "        \"log_level\": \"DEBUG\",\n",
        "        \"secret_key\": \"dev-secret-key\",\n",
        "        \"database\": {\n",
        "            \"host\": \"localhost\",\n",
        "            \"port\": 5432,\n",
        "            \"username\": \"testuser\",\n",
        "            \"password\": \"testpass\",\n",
        "            \"database\": \"testdb\"\n",
        "        },\n",
        "        \"max_workers\": 8\n",
        "    }\n",
        "\n",
        "    manager = AsyncConfigManager()\n",
        "    manager.add_dict(sample_config, priority=0)\n",
        "\n",
        "    config = await manager.load_config(AppConfig)\n",
        "\n",
        "    print(\"=== Simple Configuration Demo ===\")\n",
        "    print(f\"Debug mode: {config.debug}\")\n",
        "    print(f\"Log level: {config.log_level}\")\n",
        "    print(f\"Database host: {config.database.host}\")\n",
        "    print(f\"Database port: {config.database.port}\")\n",
        "    print(f\"Max workers: {config.max_workers}\")\n",
        "\n",
        "    return config"
      ],
      "metadata": {
        "id": "DbmIDLfegIMd"
      },
      "execution_count": 14,
      "outputs": []
    },
    {
      "cell_type": "code",
      "source": [
        "async def demo_advanced_config():\n",
        "    \"\"\"Demo advanced configuration with multiple sources.\"\"\"\n",
        "\n",
        "    base_config = {\n",
        "        \"debug\": False,\n",
        "        \"log_level\": \"INFO\",\n",
        "        \"secret_key\": \"production-secret\",\n",
        "        \"max_workers\": 4\n",
        "    }\n",
        "\n",
        "    override_config = {\n",
        "        \"debug\": True,\n",
        "        \"log_level\": \"DEBUG\",\n",
        "        \"database\": {\n",
        "            \"host\": \"dev-db.example.com\",\n",
        "            \"port\": 5433\n",
        "        }\n",
        "    }\n",
        "\n",
        "    env_config = {\n",
        "        \"secret_key\": \"env-secret-key\",\n",
        "        \"redis_url\": \"redis://prod-redis:6379\"\n",
        "    }\n",
        "\n",
        "    print(\"\\n=== Advanced Configuration Demo ===\")\n",
        "\n",
        "    manager = AsyncConfigManager()\n",
        "\n",
        "    manager.add_dict(base_config, priority=0)\n",
        "    manager.add_dict(override_config, priority=50)\n",
        "    manager.add_dict(env_config, priority=100)\n",
        "\n",
        "    config = await manager.load_config(AppConfig)\n",
        "\n",
        "    print(\"Configuration sources merged:\")\n",
        "    print(f\"Debug mode: {config.debug} (from override)\")\n",
        "    print(f\"Log level: {config.log_level} (from override)\")\n",
        "    print(f\"Secret key: {config.secret_key} (from env)\")\n",
        "    print(f\"Database host: {config.database.host} (from override)\")\n",
        "    print(f\"Redis URL: {config.redis_url} (from env)\")\n",
        "\n",
        "    return config\n",
        "\n",
        "\n",
        "async def demo_validation():\n",
        "    \"\"\"Demo configuration validation.\"\"\"\n",
        "\n",
        "    print(\"\\n=== Configuration Validation Demo ===\")\n",
        "\n",
        "    valid_config = {\n",
        "        \"debug\": True,\n",
        "        \"log_level\": \"DEBUG\",\n",
        "        \"secret_key\": \"test-key\",\n",
        "        \"database\": {\n",
        "            \"host\": \"localhost\",\n",
        "            \"port\": 5432\n",
        "        }\n",
        "    }\n",
        "\n",
        "    manager = AsyncConfigManager()\n",
        "    manager.add_dict(valid_config, priority=0)\n",
        "\n",
        "    try:\n",
        "        config = await manager.load_config(AppConfig)\n",
        "        print(\"✓ Valid configuration loaded successfully\")\n",
        "        print(f\"  Database SSL: {config.database.ssl_enabled} (default value)\")\n",
        "        print(f\"  Database pool size: {config.database.pool_size} (default value)\")\n",
        "    except ValidationError as e:\n",
        "        print(f\"✗ Validation error: {e}\")\n",
        "\n",
        "    incomplete_config = {\n",
        "        \"debug\": True,\n",
        "        \"log_level\": \"DEBUG\"\n",
        "    }\n",
        "\n",
        "    manager2 = AsyncConfigManager()\n",
        "    manager2.add_dict(incomplete_config, priority=0)\n",
        "\n",
        "    try:\n",
        "        config2 = await manager2.load_config(AppConfig)\n",
        "        print(\"✓ Configuration with defaults loaded successfully\")\n",
        "        print(f\"  Secret key: '{config2.secret_key}' (default empty string)\")\n",
        "    except ValidationError as e:\n",
        "        print(f\"✗ Validation error: {e}\")"
      ],
      "metadata": {
        "id": "R6pQiPyygUmD"
      },
      "execution_count": 15,
      "outputs": []
    },
    {
      "cell_type": "code",
      "execution_count": 8,
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "hMrV2ULodDkI",
        "outputId": "905b056f-c9cd-47b4-9740-87ae3a874129"
      },
      "outputs": [
        {
          "output_type": "stream",
          "name": "stdout",
          "text": [
            "=== Simple Configuration Demo ===\n",
            "Debug mode: True\n",
            "Log level: DEBUG\n",
            "Database host: localhost\n",
            "Database port: 5432\n",
            "Max workers: 8\n",
            "\n",
            "=== Advanced Configuration Demo ===\n",
            "Configuration sources merged:\n",
            "Debug mode: True (from override)\n",
            "Log level: DEBUG (from override)\n",
            "Secret key: env-secret-key (from env)\n",
            "Database host: dev-db.example.com (from override)\n",
            "Redis URL: redis://prod-redis:6379 (from env)\n",
            "\n",
            "=== Configuration Validation Demo ===\n",
            "✓ Valid configuration loaded successfully\n",
            "  Database SSL: False (default value)\n",
            "  Database pool size: 10 (default value)\n",
            "✓ Configuration with defaults loaded successfully\n",
            "  Secret key: '' (default empty string)\n",
            "\n",
            "=== All demos completed successfully! ===\n",
            "Running in Jupyter/IPython environment\n",
            "Use: await run_demos()\n"
          ]
        }
      ],
      "source": [
        "async def run_demos():\n",
        "    \"\"\"Run all demonstration functions.\"\"\"\n",
        "    try:\n",
        "        await demo_simple_config()\n",
        "        await demo_advanced_config()\n",
        "        await demo_validation()\n",
        "        print(\"\\n=== All demos completed successfully! ===\")\n",
        "    except Exception as e:\n",
        "        print(f\"Demo error: {e}\")\n",
        "        import traceback\n",
        "        traceback.print_exc()\n",
        "\n",
        "\n",
        "\n",
        "await run_demos()\n",
        "\n",
        "if __name__ == \"__main__\":\n",
        "    try:\n",
        "        loop = asyncio.get_event_loop()\n",
        "        if loop.is_running():\n",
        "            print(\"Running in Jupyter/IPython environment\")\n",
        "            print(\"Use: await run_demos()\")\n",
        "        else:\n",
        "            asyncio.run(run_demos())\n",
        "    except RuntimeError:\n",
        "        asyncio.run(run_demos())"
      ]
    }
  ]
}